#include "db/dbenv_impl.hxx"
#include "util/fs.hxx"
#include "util/lockguard.hxx"
#include "db/dbpartition.hxx"
#include "db/datastore_impl.hxx"
#include "db/compact_impl.hxx"
#include "dbt.hxx"
#include "util/common.hxx"
#include "util/logger.hxx"
#include <errno.h>


DataStoreImpl::DataStoreImpl(DbEnvImpl& env)
: m_env(&env), 
  m_partitions(NULL), 
  m_readOnly(false) {

}

DataStoreImpl::~DataStoreImpl() {

    if(m_partitions) close();
}


DbStatus DataStoreImpl::open(const std::string& dsName,
                         uint32_t cacheSize,
                         uint32_t indexArrayCapacity,
                         uint32_t segCapacity,
                         MAPPINGFUNC* mapFunc,
                         uint32_t flag) {

    DbStatus ret;

    int rc = pthread_rwlock_init(&m_rwlockMap, NULL);
    if(rc != 0) return DbStatus("DataStoreImpl::open::pthread_rwlock_init", rc);

	m_dsPath = m_env->getHomepath(); m_dsPath += "/" + dsName;
	if(!FsUtility::fileExists(m_dsPath)) {
	
		ret = FsUtility::createDir(m_dsPath);
        if(!ret.isok()) { 
            Logger::log("create datastore dir: %s failed, error msg:", m_dsPath.c_str(),
                    ret.msg().c_str());
            return ret;
        }
        Logger::log("create datastore dir: %s", m_dsPath.c_str());
	}

	if(!FsUtility::isDir(m_dsPath)) {
	
		return DbStatus(m_dsPath, EINVAL);
	}

	if(indexArrayCapacity < 4096) return DbStatus("DataStore::open indexArrayCapacity", EINVAL);

	if(mapFunc == NULL) return DbStatus("DataStore::open", EINVAL);
	
	m_dsname = 	dsName;
	m_cacheSize = cacheSize;
	m_mapFunc = mapFunc;
	m_indexArrayCapacity = indexArrayCapacity;
	m_segmentCapacity = segCapacity;
    m_readOnly = (flag & DATASTORE_READONLY)? true:false; 
    m_flag = flag;

	ret = init();

    m_env->addStore(this);
    m_partitions = new std::map<std::string, DbPartition*>();

    Logger::log("open datastore %s successfully, cache size: %d, index array \
            capacity: %d, segment capacity: %d, flag: %d, readonly: %s", 
            m_dsname.c_str(),
            m_cacheSize,
            m_indexArrayCapacity,
            m_segmentCapacity,
            m_flag,
            m_readOnly?"yes":"no");

    return ret;
}

DbStatus DataStoreImpl::close() {

    m_env->stopCompact(true);

    RWLockGuard lock(m_rwlockMap, 'w');

    for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
            iter != m_partitions->end(); ++iter) {

        if(iter->second) delete iter->second;
    }

    m_partitions->clear();
    lock.reset();

    if(m_partitions) {delete m_partitions; m_partitions = NULL; }
    pthread_rwlock_destroy(&m_rwlockMap);


    Logger::log("close datastore %s successfully", m_dsname.c_str());
    
	return DbStatus("", 0);
}

DbStatus DataStoreImpl::init() {

	std::vector<std::string> children;
	FsUtility::getChildren(m_dsPath, children);
	for(int i = 0; i < children.size(); ++i) {
	
		std::string partitionPath = m_env->getHomepath(); partitionPath += "/" + children[i];
		if(FsUtility::isDir(partitionPath)) {
		
			DbPartition* dbPart = new DbPartition(*this, *m_env->getLogManager(), children[i]);
			DbStatus ret = dbPart->init();
			if(!ret.isok()) {delete dbPart; return ret;}
			(*m_partitions)[children[i]] = dbPart;
		}
	}

	return DbStatus("DataStore::init", 0);
}

DbStatus DataStoreImpl::flush() {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'r');

	for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
		 iter != m_partitions->end(); ++iter) {
	
		ret = iter->second->flush(); if(!ret.isok())  return ret;
	}

	return ret; 
}


DbStatus DataStoreImpl::flushSegments() {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'r');

    for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
            iter != m_partitions->end(); ++iter) {

        ret = iter->second->flushSegments(); if(!ret.isok())  return ret;
    }

    return ret;
}

DbStatus DataStoreImpl::flushIndexArray() {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'r');

    for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
            iter != m_partitions->end(); ++iter) {

        ret = iter->second->flushIndexArray(); if(!ret.isok())  return ret;
    }

    return ret;
}


DbStatus DataStoreImpl::get(Dbt& key, Dbt& value) {

    Dbt dbt(key);
	std::string partionName = m_mapFunc(dbt);

	DbPartition* part = findPartition(partionName);
    if(part == NULL) return DbStatus("DataStoreImpl::get", DbStatus::DB_PARTITIONNOTFOUND);

	return part->get(key, value);
}


DbStatus DataStoreImpl::put(Dbt& key, Dbt& value) {

    if(m_readOnly) return DbStatus("DataStoreImpl::put", DbStatus::DB_READONLY);

    std::string partionName = m_mapFunc(key);

    DbPartition* part = findPartition(partionName);
    if(part == NULL) return DbStatus("DataStoreImpl::put", DbStatus::DB_PARTITIONNOTFOUND);

	DbStatus ret = part->put(key, value);

    return ret;
}


DbStatus DataStoreImpl::del(Dbt& key) {

    if(m_readOnly) return DbStatus("DataStoreImpl::put", DbStatus::DB_READONLY);

    std::string partionName = m_mapFunc(key);

    DbPartition* part = findPartition(partionName);
    if(part == NULL) return DbStatus("DataStoreImpl::del", DbStatus::DB_PARTITIONNOTFOUND);

    return part->del(key);
}


DbStatus DataStoreImpl::addPartition(const std::string& partitionName) {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'w');

    if(m_partitions->count(partitionName) > 0) return DbStatus("addPartition", 0);
        
    std::string partPath = m_dsPath + "/" + partitionName;
    if(!FsUtility::fileExists(partPath)) {
    
        ret = FsUtility::createDir(partPath); if(!ret.isok()) return ret;
    }

    if(!FsUtility::isDir(partPath)) {
    
        return DbStatus(partPath, EINVAL);
    }

    DbPartition* dbPart = new DbPartition(*this, *m_env->getLogManager(), partitionName);
    ret = dbPart->init();
    if(!ret.isok()) {delete dbPart; return ret;}


    (*m_partitions)[partitionName] = dbPart;

    Logger::log("add partition: %s", partitionName.c_str());

    return ret;
}

DbStatus DataStoreImpl::removePartition(const std::string& partitionName) {
    
    RWLockGuard lock(m_rwlockMap, 'w');

    if(m_partitions->count(partitionName) == 0) return DbStatus("removePartition", DbStatus::DB_PARTITIONNOEXIST);

    m_partitions->erase(partitionName);

    Logger::log("remove partition: %s", partitionName.c_str());
}


DbPartition* DataStoreImpl::findPartition(const std::string& partitionName) {

    DbPartition* part = NULL;

    RWLockGuard lock(m_rwlockMap, 'r');

    std::map<std::string, DbPartition*>::iterator iter =
        m_partitions->find(partitionName);

    if(iter != m_partitions->end()) part = iter->second;

    return part;
}


DbStatus DataStoreImpl::compact(CompactOption& option) {

   DbStatus ret;

   for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
            iter != m_partitions->end(); ++iter) {

        if(iter->second) ret = iter->second->compact(option);
        if(!ret.isok()) break;
    }

   return ret;
}

//protected by sync lock
void DataStoreImpl::switchEntries() {
       
   for(std::map<std::string, DbPartition*>::iterator iter = m_partitions->begin();
            iter != m_partitions->end(); ++iter) {
        if(iter->second) iter->second->switchEntries();
    }

}

